package com.example.cdc.server;

import com.example.cdc.constant.CDCConstant;
import com.example.cdc.listener.CDCConsoleListener;
import com.example.cdc.listener.CDCListener;
import com.example.po.SysCdcListener;
import com.example.utils.JdbcUtils;
import com.example.cdc.wrap.BinlogAddObj;
import com.example.cdc.wrap.BinlogDeleteObj;
import com.example.cdc.wrap.BinlogObj;
import com.example.cdc.wrap.BinlogUpdateObj;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.Serializable;
import java.math.BigInteger;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Data
public class CDCServer {

    Logger logger = LoggerFactory.getLogger(CDCServer.class);

    private Map<Long, String> tableMap = new ConcurrentHashMap<>();

    private Map<String, Map<Long, String>> colMap = new ConcurrentHashMap<>();

    private String host;

    private int port;

    private String db;

    private String username;

    private String password;

    private String driverClassName;

    private int serverId;

    private JdbcUtils jdbcUtils;

    private BinaryLogClient binaryLogClient;

    private ExecutorService executorService;

    private CDCListener cdcListener;

    private SysCdcListener sysCdcListener;

    private RedisTemplate<String, Object> redisTemplate;

    private BinaryLogClient.EventListener eventListener = new BinaryLogClient.EventListener() {
        @Override
        public void onEvent(Event event) {
            if (event.getData() instanceof RotateEventData) {
                RotateEventData rotateEventData = event.getData();
                redisTemplate.opsForValue().getAndSet(CDCConstant.REDIS_PATH_PREF + serverId, rotateEventData.getBinlogFilename());
                redisTemplate.opsForValue().getAndSet(CDCConstant.REDIS_INDEX_PREF + serverId, rotateEventData.getBinlogPosition());
            } else {
                EventHeaderV4 header = event.getHeader();
                redisTemplate.opsForValue().getAndSet(CDCConstant.REDIS_INDEX_PREF + serverId, header.getPosition());
            }
            if (event.getData() instanceof TableMapEventData) {
                TableMapEventData eventData = event.getData();
                long tableId = eventData.getTableId();
                String tableName = eventData.getTable();
                tableMap.putIfAbsent(tableId, tableName);
            } else {
                BinlogObj binlogObj = wrapBinlogObj(event);
                if (binlogObj != null) {
                    executorService.execute(new CDCServerThread(binlogObj));
                }
            }
        }
    };

    private BinaryLogClient.LifecycleListener lifecycleListener = new BinaryLogClient.LifecycleListener() {

        @Override
        public void onConnect(BinaryLogClient client) {
            logger.warn("Connected to MySQL server: " + client.getConnectionId());
        }

        @Override
        public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
            logger.warn("Communication failure: " + ex.getMessage());
        }

        @Override
        public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
            logger.warn("Event deserialization failure: " + ex.getMessage());
        }

        @Override
        public void onDisconnect(BinaryLogClient client) {
            logger.warn("Disconnected from MySQL server: " + client.getConnectionId());
        }
    };

    public void init() {
        Map<Long, String> colMap = new HashMap<>();
        List<Map<String, Object>> maps = jdbcUtils.executeQuery("select ORDINAL_POSITION ,COLUMN_NAME from information_schema.COLUMNS where TABLE_SCHEMA = '" + db + "' and TABLE_NAME = '" + sysCdcListener.getTableName() + "'");
        for (Map<String, Object> map : maps) {
            if (map.containsKey(CDCConstant.ORDINAL_POSITION) && map.containsKey(CDCConstant.COLUMN_NAME)) {
                BigInteger ordinalPosition = (BigInteger) map.get(CDCConstant.ORDINAL_POSITION);
                colMap.put(ordinalPosition.longValue(), (String) map.get(CDCConstant.COLUMN_NAME));
            }
        }
        if (!colMap.isEmpty()) {
            this.colMap.put(sysCdcListener.getTableName(), colMap);
        }
    }

    public void start() {
        this.executorService = Executors.newCachedThreadPool();
        this.executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    binaryLogClient = new BinaryLogClient(host, port, db, username, password);
                    binaryLogClient.setServerId(serverId);
                    binaryLogClient.registerEventListener(eventListener);
                    binaryLogClient.registerLifecycleListener(lifecycleListener);
                    if (redisTemplate.hasKey(CDCConstant.REDIS_PATH_PREF + serverId) && redisTemplate.hasKey(CDCConstant.REDIS_INDEX_PREF + serverId)) {
                        binaryLogClient.setBinlogFilename((String) redisTemplate.opsForValue().get(CDCConstant.REDIS_PATH_PREF + serverId));
                        binaryLogClient.setBinlogPosition((Long) redisTemplate.opsForValue().get(CDCConstant.REDIS_INDEX_PREF + serverId));
                    }
                    binaryLogClient.connect();
                } catch (Exception e) {
                    logger.warn("start cdc server error: " + e.getMessage());
                }
            }
        });
    }

    public void stop() throws Exception {
        this.binaryLogClient.disconnect();
        this.executorService.shutdownNow();
    }

    private BinlogObj wrapBinlogObj(Event event) {
        BinlogObj binlogObj = null;
        if (event.getData() instanceof UpdateRowsEventData) {
            EventHeaderV4 header = event.getHeader();
            UpdateRowsEventData updateRowsEventData = event.getData();
            String tableName = this.tableMap.get(updateRowsEventData.getTableId());
            if (!tableName.equalsIgnoreCase(sysCdcListener.getTableName())) {
                return null;
            }
            binlogObj = new BinlogUpdateObj(db, tableName, CDCConstant.UPDATE, header.getTimestamp(), updateRowsEventData);
            Map<Long, String> cols = this.colMap.get(tableName);
            for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                Map<String, Map<Serializable, Serializable>> colMap = new HashMap<>();
                Serializable[] key = row.getKey();
                Serializable[] value = row.getValue();
                for (int i = 0; i < key.length; i++) {
                    Map<Serializable, Serializable> valMap = new HashMap<>();
                    String colName = cols.get((long) (i + 1));
                    valMap.put(key[i], value[i]);
                    colMap.put(colName, valMap);
                }
                ((BinlogUpdateObj) binlogObj).getUpdateRowList().add(colMap);
            }
        }
        if (event.getData() instanceof WriteRowsEventData) {
            EventHeaderV4 header = event.getHeader();
            WriteRowsEventData writeRowsEventData = event.getData();
            String tableName = this.tableMap.get(writeRowsEventData.getTableId());
            if (!tableName.equalsIgnoreCase(sysCdcListener.getTableName())) {
                return null;
            }
            binlogObj = new BinlogAddObj(db, tableName, CDCConstant.INSERT, header.getTimestamp(), event.getData());
            Map<Long, String> cols = this.colMap.get(tableName);
            BitSet includedColumns = writeRowsEventData.getIncludedColumns();
            for (Serializable[] row : writeRowsEventData.getRows()) {
                Map<String, Serializable> map = new HashMap<>();
                for (int i = 0; i < includedColumns.length(); i++) {
                    int colIndex = includedColumns.nextSetBit(i);
                    Long colIndexLong = (long) (colIndex + 1);
                    map.put(cols.get(colIndexLong), row[colIndex]);
                }
                ((BinlogAddObj) binlogObj).getRowsList().add(map);
            }
        }
        if (event.getData() instanceof DeleteRowsEventData) {
            EventHeaderV4 header = event.getHeader();
            DeleteRowsEventData deleteRowsEventData = event.getData();
            String tableName = this.tableMap.get(deleteRowsEventData.getTableId());
            if (!tableName.equalsIgnoreCase(sysCdcListener.getTableName())) {
                return null;
            }
            binlogObj = new BinlogDeleteObj(db, tableName, CDCConstant.DELETE, header.getTimestamp(), event.getData());
            Map<Long, String> cols = this.colMap.get(tableName);
            BitSet includedColumns = deleteRowsEventData.getIncludedColumns();
            List<Serializable[]> rows = deleteRowsEventData.getRows();
            for (Serializable[] row : rows) {
                Map<String, Serializable> map = new HashMap<>();
                for (int i = 0; i < includedColumns.length(); i++) {
                    int colIndex = includedColumns.nextSetBit(i);
                    Long colIndexLong = (long) (colIndex + 1);
                    map.put(cols.get(colIndexLong), row[colIndex]);
                }
                ((BinlogDeleteObj) binlogObj).getRowsList().add(map);
            }
        }
        return binlogObj;
    }

    public CDCServer(SysCdcListener sysCdcListener, RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.sysCdcListener = sysCdcListener;
        this.host = sysCdcListener.getHost();
        this.port = sysCdcListener.getPort();
        this.db = sysCdcListener.getSchemaName();
        this.username = sysCdcListener.getUserName();
        this.password = sysCdcListener.getPassWord();
        this.driverClassName = sysCdcListener.getDriverClassName();
        this.serverId = sysCdcListener.getId().intValue();
        this.cdcListener = new CDCConsoleListener();
        this.jdbcUtils = new JdbcUtils(this.host, this.port, this.db, this.username, this.password, this.driverClassName);
    }

    public CDCServer(SysCdcListener sysCdcListener, CDCListener cdcListener, RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.cdcListener = cdcListener;
        this.sysCdcListener = sysCdcListener;
        this.host = sysCdcListener.getHost();
        this.port = sysCdcListener.getPort();
        this.db = sysCdcListener.getSchemaName();
        this.username = sysCdcListener.getUserName();
        this.password = sysCdcListener.getPassWord();
        this.driverClassName = sysCdcListener.getDriverClassName();
        this.serverId = sysCdcListener.getId().intValue();
        this.cdcListener = new CDCConsoleListener();
        this.jdbcUtils = new JdbcUtils(this.host, this.port, this.db, this.username, this.password, this.driverClassName);
    }

    class CDCServerThread implements Runnable {

        private BinlogObj binlogObj;

        public CDCServerThread(BinlogObj binlogObj) {
            this.binlogObj = binlogObj;
        }

        @Override
        public void run() {
            cdcListener.onMessage(binlogObj);
        }
    }
}
